// voter-server.c -- Read clients input, vote and give result as output.
//=========================================================================
// Copyright (C) 2024 The C++ Component Model(COMO) Open Source Project
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//=========================================================================

#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <assert.h>
#include <sys/select.h>
#include <unistd.h>
#include <math.h>
#include <time.h>
#include <syslog.h>
#include <inttypes.h>
#include <stdint.h>
#include "zmtp.h"
#include "int64.h"
#include "lamport_clock.h"
#include "math_utils.h"
#include "checksum.h"
#include "voter_message.h"
#include "voter_lib.h"
#include "voter_server.h"
#include "ZMQ_Switcher.h"
#include "voter_logger.h"
#include "config.h"
#include "rudp.h"
#include "voter_VotingAlgorithms.h"
#include "voter_customized.h"
#ifdef SYNC_WITH_PHXPAXOS
#include "PhxPaxosC.h"
#endif

#define max(a,b) ((a) >= (b) ? (a) : (b))

static ChatState Chat;      // Initialized at startup.
ChatState *p_Chat = &Chat;
LamportClock g_LamportClock = {0};

extern Rudp rudp;
// We don't define a global variable msg_recv like rudp/test/server.c did,
// since a local variable readbuf is defined (see handle_udp_data())

int g_iSERVER_PORT = SERVER_PORT;
int g_iSERVER_PORT_UDP = SERVER_PORT_UDP;
char g_iSERVER_NAMESPACE[LINUX_NAMESPACE_LEN] = {'\0'};
char g_iSERVER_NAMESPACE_UDP[LINUX_NAMESPACE_LEN] = {'\0'};

#ifndef SYNC_WITH_PHXPAXOS
// 如果没有 Paxos，则这个函数需要自己提供一个空的
int PhxIsIMMaster(void *oPaxosServer)
{
    (void)oPaxosServer;

    return 0;
}
#endif

/**
 * Voter core implementation
 * Here the idea is very simple: we accept new connections, read what clients
 * write us and fan-out (that is, send-to-all) the message to everybody
 * with the exception of the sender. And that is, of course, the most
 * simple chat system ever possible.
 */

/**
 * Create a new client bound to 'fd'. This is called when a new client
 * connects. As a side effect updates the global Chat state.
 */
VoterClient *CreateClient(int fd, struct sockaddr_in *sa)
{
    int i;

    // find the first empty slot(i.e. fd==-1).
    // i+1 is the num of clients.
    // That is, the 1st client would be p_Chat->clients[0].
    //          the 2nd client would be p_Chat->clients[1], etc.
    //    suppose the current client is p_Chat->clients[i].
    for (i = 0;  i < MAX_CLIENTS;  i++) {
        if (-1 == p_Chat->clients[i].fd) {
            break;
        }
    }
    // when num of clients reaches limit, the creation fails.
    if (i >= MAX_CLIENTS) {
        WriteLog("clients reaches limit, the creation fails\n", -1);
        return NULL;
    }

    // define a pointer to p_Chat->clients[i]
    VoterClient *c = &(p_Chat->clients[i]);

    SocketSetNonBlockNoDelay(fd);   // Pretend this will not fail.
    c->fd = fd;
    c->idxDealer = -1;

    // Extract the IP address
    inet_ntop(AF_INET, &(sa->sin_addr), c->client_ip, INET_ADDRSTRLEN);

    // Extract the port number and convert it from network endianness to host endianness
    c->client_port = ntohs(sa->sin_port);

    // We need to update the max client set if needed.
    if (c->fd > p_Chat->max_client_fd) {
        p_Chat->max_client_fd = c->fd;
    }
    p_Chat->numclients++;

    return c;
}

/**
 * Free a client, associated resources, and unbind it from the global
 * state in Chat.
 */
int FreeClient(int clientId)
{
    int i;
    VoterClient *c;

    if ((clientId >= 0) && (clientId < MAX_CLIENTS)) {
        c = &(p_Chat->clients[clientId]);
        close(c->fd);
        c->fd = -1;
    }
    else {
        return -1;
    }

    p_Chat->numclients--;
    if (p_Chat->max_client_fd == c->fd) {
        /**
         * This was the max client set. Let's find what is
         * the new highest slot used.
         */
        for (i = p_Chat->max_client_fd - 1;  i >= 0;  i--) {
            if (-1 != p_Chat->clients[i].fd) {
                p_Chat->max_client_fd = i;
                break;
            }
        }
        if (-1 == i) {
            p_Chat->max_client_fd = -1;     // We no longer have clients.
        }
    }

    return clientId;
}

/**
 * Allocate and init the global stuff.
 */
void InitVoter(void)
{
    memset(p_Chat, '\0', sizeof(ChatState));

    for (int i = 0;  i < MAX_CLIENTS;  i++) {
        p_Chat->clients[i].fd = -1;
        p_Chat->clients[i].idxDealer = -1;
    }

    // No clients at startup, of course.
    p_Chat->max_client_fd = -1;
    p_Chat->numclients = 0;
    p_Chat->cur_trans_id = 0;
    p_Chat->cur_idTransaction = 0;
    p_Chat->cur_numVoters = 0;
    p_Chat->iKeepPaused = 0;

    p_Chat->pVoterOutputMsg = g_pconfig->voter_OUTPUT_MSG;

    /**
     * Create our listening socket, bound to the given port. This
     * is where our clients will connect.
     */
    p_Chat->serversock = CreateTCPServer(g_iSERVER_PORT, g_iSERVER_NAMESPACE);
    if (-1 == p_Chat->serversock) {
        printf("Failed creating TCP listening socket\n");
        exit(1);
    }

    p_Chat->serversockUdp = CreateUDPServer(g_iSERVER_PORT_UDP, g_iSERVER_NAMESPACE_UDP);
    if (-1 == p_Chat->serversockUdp) {
        printf("Failed creating UDP listening socket\n");
        exit(1);
    }

    pthread_mutex_init(&(p_Chat->mutex), NULL);
}

/**
 * CleanIdleClient
 */
void CleanIdleClient(void)
{
    struct timespec now;
    clock_gettime(CLOCK_REALTIME, &now);

    for (int i = 0;  i < MAX_CLIENTS;  i++) {
        VoterClient *c = &(p_Chat->clients[i]);

        if (c->fd < 0) {
            continue;
        }

        int64_t timeL = now.tv_sec + now.tv_nsec / 1000000000 -
                       (c->lastAccessTime.tv_sec+ //123456789        123456789
                                        c->lastAccessTime.tv_nsec / 1000000000);

        if (timeL > g_pconfig->max_CLIENT_IDLE_TIME) {
            struct tcp_info info;
            int len = sizeof(info);

            p_Chat->max_client_fd = 0;

            // Determine whether the socket(p_Chat->clients[i].fd) is disconnected
            getsockopt(c->fd, IPPROTO_TCP, TCP_INFO, &info, (socklen_t *)&len);
            if (info.tcpi_state != TCP_ESTABLISHED)  {
                if (c->idxDealer >= 0) {
                    zmtp_voter_destroy(c->idxDealer);
                }
                else {
                    close(c->fd);
                }
                c->fd = -1;
            }
            else {
                if (c->fd > p_Chat->max_client_fd) {
                    p_Chat->max_client_fd = c->fd;
                }
            }
        }
    } // end of for
}

/**
 * Send the specified string to all connected clients but the one
 * having as socket descriptor 'excluded'. If you want to send something
 * to every client just set excluded to an impossible socket: -1.
 */
void SendMsgToAllClients(int excluded, char *s, size_t len)
{
    for (int i = 0;  i <= p_Chat->max_client_fd;  i++) {
        int fdTmp = p_Chat->clients[i].fd;

        // fd == 0 means the server itself, so exclude
        if ((fdTmp <= 0) || (fdTmp == excluded)) {
            continue;
        }

        if (p_Chat->clients[i].idxDealer > 0) {
            zmtp_voter_send(p_Chat->clients[i].idxDealer, s, len);
            continue;
        }

        /**
         * Important: we don't do ANY BUFFERING. We just use the kernel
         * socket buffers. If the content does not fit, we don't care.
         * This is needed in order to keep this program simple.
         */
        size_t num = write(fdTmp, s, len);
        if (num != len) {
            WriteLog("SendMsgToAllClients write error\n", -1);
            continue;
        }
    }
}

/**
 * 表决结果输出
 *
 * 1. g_voterOutputTokenOrder 在全局变量存储区, outputTokenOrder在栈上，通过两个渠
 *    道传入的序号，得到token，与栈上传入的 token 一致，方可对外输出，防护程序执行路径
 *    错误而引起的错误输出。
 * 2. ...
 */
int VoterOutputMsg(char *s, size_t len, int outputTokenOrder, uint64_t token,
                                                             Transaction *trans)
{
    (void)s;
    (void)len;
    (void)trans;

    if ((outputTokenOrder < 0) || (outputTokenOrder >= g_numVoterOutputToken)) {
        return -1;
    }

    if (g_voterOutputTokenOrder != outputTokenOrder) {
        return -2;
    }

    if (g_i64bit_Tokens[g_voterOutputTokenOrder] != token) {
        return -3;
    }

    // Output the voting results

    return 0;
}

/**
 * ExtractData函数
 * 从客户端消息中提取数据，并根据条件更新事务中的数据状态
 *
 * @param clientId 客户端ID，用于索引事务数据数组
 * @param message  指向VoterMessage结构体的指针，包含客户端发送的消息
 * @param trans    指向Transaction结构体的指针，代表当前处理的事务
 * @return         返回0表示成功处理并更新数据，返回-1表示未找到对应的事务
 */
int ExtractData(int clientId, VoterMessage *message)
{
    Transaction *trans;

    // 如果message给出的事务ID是当前事务的ID
    if (message->idTransaction == p_Chat->cur_idTransaction) {

        // 防护一下，免得数据访问越界
        if ((p_Chat->cur_trans_id < 0) || (p_Chat->cur_trans_id > MAX_NUM_TRANS)) {
            syslog(LOG_ERR, "ExtractData, p_Chat->cur_idTransaction: Bad data");
            return -1;
        }

        trans = &(p_Chat->trans_arr[p_Chat->cur_trans_id]);

        // 一个客户端多次提交同一表决数据，取其最后一次数据有效
        trans->data[clientId].i64Val = message->data;
        //memcpy(&(trans->data[clientId]), message, message->len);

        // 如果数据状态为NULL，表示第一次接收到该客户端的数据，更新数据状态为READY
        if (DATA_NULL == trans->dataState[clientId]) {
            p_Chat->current_data_num++;
            trans->cur_data_num++;
            trans->dataState[clientId] = DATA_READY;
        }

        return 0;
    }
    // 如果不是，遍历所有事务，查找是否有匹配的事务ID
    else {
        for (int i = 0;  i < MAX_NUM_TRANS;  i++) {
            if (p_Chat->trans_arr[i].idTransaction == message->idTransaction) {
                // 提交非当前表决数据，取其最后一次数据有效
                trans = &(p_Chat->trans_arr[i]);

                trans->data[clientId].i64Val = message->data;
                //memcpy(&(trans->data[clientId]), message, message->len);

                // 如果数据状态为NULL，表示第一次接收到该客户端的数据，更新数据状态为READY
                if (DATA_NULL == trans->dataState[clientId]) {
                    p_Chat->current_data_num++;
                    trans->cur_data_num++;
                    trans->dataState[clientId] = DATA_READY;
                }
                return 0;
            }
        }
    }

    /**
     * 客户端请求数据，事务ID是新的，表示新的一轮表决
     */
    if (! ValidIdTransaction(message->idTransaction)) {
        return -2;
    }

    // 循环使用 p_Chat->trans_arr 中的空间
    p_Chat->cur_trans_id = (p_Chat->cur_trans_id + 1) % MAX_NUM_TRANS;

    trans = &(p_Chat->trans_arr[p_Chat->cur_trans_id]);

    // 一个客户端多次提交同一表决数据，取其最后一次数据有效
    trans->data[clientId].i64Val = message->data;
    //memcpy(&(trans->data[clientId]), message, message->len);

    /**
     * 要表决的事务是新的，但表决周期不是新的
     * XXX p_Chat->current_data_num = 1;
     */
    trans->cur_data_num = 1;
    trans->dataState[clientId] = DATA_READY;

    p_Chat->current_data_num++;
    p_Chat->cur_idTransaction = message->idTransaction;
    trans->idTransaction = message->idTransaction;
    trans->votingResult = VOTING_NULL;

#ifdef SYNC_WITH_PHXPAXOS
    for (int i = 0;  i < p_Chat->cur_numVoters;  i++) {
        char keybuf[MAX_ID_SIZE + 64];

        snprintf(keybuf, sizeof(keybuf), MACRO_VR_PRINTF_FMTSTR,
                                     p_Chat->idVoters[i], trans->idTransaction);
        // 尝试性删除表决结果信息（VResult#--#--），不用判断返回值
        (void)PhxLevelDbDelete(keybuf);
    }
#endif

    return 0;
}

/**
 * ExtractBigData
 */
int ExtractBigData(int clientId, VoterMessage *message)
{
    Transaction *trans;

    if (message->len > MAX_VoterMessage_SIZE) {
        return -1;
    }

#ifdef __DBEUG__
    printf("ExtractBigData: clientId: %d, message->idTransaction: %" PRIu64 ", "
           "message->len: %d, p_Chat->cur_idTransaction: %" PRId64 "\n", clientId,
           message->idTransaction, message->len, (int64_t)p_Chat->cur_idTransaction);
#endif

    // 仅当client给出当前待判决事务的id时，才算增加一条数据，更新current_data_num
    if (message->idTransaction == p_Chat->cur_idTransaction) {
        // 一个客户端多次提交同一表决数据，取其最后一次数据有效
        // 仅把buf部分（存储bigdata数据）以及crc32保存下来

        // 防护一下，免得数据访问越界
        if ((p_Chat->cur_trans_id < 0) || (p_Chat->cur_trans_id > MAX_NUM_TRANS)) {
            syslog(LOG_ERR, "ExtractBigData, p_Chat->cur_idTransaction: Bad data");
            return -2;
        }

        trans = &(p_Chat->trans_arr[p_Chat->cur_trans_id]);
        if (DATA_NULL == trans->dataState[clientId]) {
            memcpy(&(trans->bigVotingData[clientId]), message->buf,
                                            message->len - VoterMessage_HEAD_LEN);

            p_Chat->current_data_num++;
            trans->cur_data_num++;
            trans->dataState[clientId] = DATA_READY;

            // For big data, we do not use i64val
            trans->data[clientId].u64Val = 0;
            trans->data[clientId].u32Val = message->crc32;
            trans->lenData = message->len - VoterMessage_HEAD_LEN;
            trans->isBigData = 1;
            trans->idTransaction = message->idTransaction;
        }

        return 0;
    }

    // 如果不是，遍历所有事务，查找是否有匹配的事务ID
    for (int i = 0;  i < MAX_NUM_TRANS;  i++) {
        if (p_Chat->trans_arr[i].idTransaction == message->idTransaction) {
            // 提交非当前表决数据，取其最后一次数据有效
            trans = &(p_Chat->trans_arr[i]);
            if (DATA_NULL == trans->dataState[clientId]) {
                memcpy(&(trans->bigVotingData[clientId]), message->buf,
                                            message->len - VoterMessage_HEAD_LEN);

                p_Chat->current_data_num++;
                trans->cur_data_num++;
                trans->dataState[clientId] = DATA_READY;

                // For big data, we do not use i64val
                trans->data[clientId].u64Val = 0;
                trans->data[clientId].u32Val = message->crc32;
                trans->lenData = message->len - VoterMessage_HEAD_LEN;
                trans->isBigData = 1;
                trans->idTransaction = message->idTransaction;
            }

            return 0;
        }
    }

    /**
     * 客户端请求数据，事务ID是新的，表示新的一轮表决
     */
    if (! ValidIdTransaction(message->idTransaction)) {
        return -3;
    }

    // 循环使用 p_Chat->trans_arr 中的空间
    p_Chat->cur_trans_id =  (p_Chat->cur_trans_id + 1) % MAX_NUM_TRANS;

    trans = &(p_Chat->trans_arr[p_Chat->cur_trans_id]);

    /**
     * 要表决的事务是新的，但表决周期不是新的
     * XXX p_Chat->current_data_num = 1;
     */
    p_Chat->current_data_num++;
    memcpy(&(trans->bigVotingData[clientId]), message->buf,
                                            message->len - VoterMessage_HEAD_LEN);
    trans->cur_data_num = 1;

    for (int i = 0;  i < MAX_CLIENTS;  i++) {
        trans->dataState[i] = DATA_NULL;
    }
    trans->dataState[clientId] = DATA_READY;

    // For big data, we do not use i64val
    trans->data[clientId].u64Val = 0;
    trans->data[clientId].u32Val = message->crc32;
    trans->lenData = message->len - VoterMessage_HEAD_LEN;
    trans->isBigData = 1;
    trans->idTransaction = message->idTransaction;
    trans->votingResult = VOTING_NULL;

    p_Chat->cur_idTransaction = message->idTransaction;

#ifdef SYNC_WITH_PHXPAXOS
    for (int i = 0;  i < p_Chat->cur_numVoters;  i++) {
        char keybuf[MAX_ID_SIZE + 64];

        snprintf(keybuf, sizeof(keybuf), MACRO_VR_PRINTF_FMTSTR,
                                     p_Chat->idVoters[i], trans->idTransaction);
        // 尝试性删除表决结果信息（VResult#--#--），不用判断返回值
        (void)PhxLevelDbDelete(keybuf);
    }
#endif

    return 0;
}

/**
 * vote，多算法表决，在计算上冗余，满足功能安全计算要求
 * - 表决成功，返回 数据的下标
 * - 表决不成功，返回 -1
 * - 因数据量不足，或不满足异构要求，未表决，返回 -2
 * - 表决出错，其它负数
 */
int Vote(Transaction *trans)
{
    int modeIdx_0, modeIdx_1;
    VOTING_RESULT votingResult;

    if (g_pconfig->arch_total > 0) {
        int arch_count[MAX_ARCH_SUPPORT];
        memset(&arch_count, '\0', sizeof(arch_count));

        // 节点异构计算是否满足要求计数
        for (int i = 0;  i < MAX_CLIENTS;  i++) {
            if (DATA_READY == trans->dataState[i]) {
                for (int j = 0;  j < g_pconfig->arch_total;  j++) {
                    int match_len = 0;
                    int pos;
                    const char *p = p_Chat->clients[i].idVoter;

                    while ((pos = re_matchp((re_t)&g_pconfig->archPatterns[i], p,
                                                            &match_len)) != -1) {
                        arch_count[i]++;
                        p += pos + match_len;
                    }
                }
            }
        }

        // 节点异构计算是否满足要求判断
        for (int j = 0;  j < g_pconfig->arch_total;  j++) {
            if (arch_count[j] <= 0) {
                return -2;
            }
        }
    }

    int data_count = 0;
    for (int i = 0;  i < MAX_CLIENTS;  i++) {
        if (DATA_READY == trans->dataState[i]) {
            data_count++;
        }
    }
    if (data_count < g_pconfig->M_in_NooM) {
        return -2;
    }

    // Vote of common data (not big data)
    if (! (trans->isBigData)) {
        // Use two methods to find mode, return mode's index in trans->data[] array
        modeIdx_0 = FindMode(trans->data, MAX_CLIENTS, g_pconfig->N_in_NooM,
                                                                trans->dataState);
        modeIdx_1 = MajorityElement(trans->data, MAX_CLIENTS, g_pconfig->N_in_NooM,
                                                                trans->dataState);
        // If mode is found & is same in both methods, vote success
        if ((modeIdx_0 > -1) && (modeIdx_0 == modeIdx_1)) {
#ifdef __DBEUG__
            printf("Vote success! Mode = %ld\n", trans->data[modeIdx_0].i64Val);
#endif
        }
        // Vote fail
        else {
            modeIdx_0 = -1;
#ifdef __DBEUG__
            printf("not trans->isBigData, Vote fail(no mode)!\n");
            for (int i = 0;  (i < MAX_CLIENTS) && (DATA_READY == trans->dataState[i]);  i++) {
                printf("%ld ", trans->data[i].i64Val);
            }
            printf("\n");
#endif
        }

        return modeIdx_0;
    }

    /**
     * 大数(大于 sizeof(Int64))数据比较，除了对trans->data中的crc32比较，还要
     * 检查数据本身
     *
     * 大数(大于 sizeof(Int64))数据比较，对trans->data中的crc32进行复查，一是
     * crc32可能有冲突风险，还是要对原始数据进行比较；二是这时如果出现数据失效，
     * 要对其进行定位
     */
    // 检查存储的数据，防止内存失效
    for (int i = 0;  i < MAX_CLIENTS;  i++) {
        if (DATA_READY == trans->dataState[i]) {
            uint32_t crc32 = crc_32(
                (const unsigned char *)&(trans->bigVotingData[i]), trans->lenData);

            if (trans->data[i].u32Val != crc32) {
                //TODO
                const char *msg = "error, CheckBoxVoterMessage fails.";
                if (p_Chat->monitorsockUdp > 0) {
                    if (sendto(p_Chat->monitorsockUdp, msg, strlen(msg), 0,
                                      (struct sockaddr *)&(p_Chat->monitorAddr),
                                      sizeof(p_Chat->monitorAddr)) < 0) {
                        perror("sendto");
                    }
                }

                // 数据失效了
                WriteLog("error, CheckBoxVoterMessage fails.", -1);
#ifdef __DBEUG__
                printf("error, CheckBoxVoterMessage fails. "
                       "cliendId: %d, trans->lenData: %d\n", i, trans->lenData);
#endif
                g_pVoterRunningState->crcErrorMemoryOrBus++;

                // 从PhxPaxos中取出另一份数据
#ifdef SYNC_WITH_PHXPAXOS
                uint64_t llInstanceID;
                size_t bufSize = MAX_VOTING_DATA_SIZE;
                char   buf[MAX_VOTING_DATA_SIZE];

                char   idVoterBuf[MAX_ID_SIZE + 64];
                snprintf(idVoterBuf, sizeof(idVoterBuf), MACRO_VOTERID_PRINTF_FMTSTR,
                              p_Chat->clients[i].idVoter, trans->idTransaction);

                int iRet = PhxLevelDbRead(idVoterBuf, buf, &bufSize, &llInstanceID);
                if (iRet < 0) {
                    return -3;
                }
                else {
                    // 检查 PhxPaxos 中的备份数据是否可用
                    ;
                }
#else
                return -3;
#endif
            }
        }
    }

    // 先比较一下它们的CRC校验值(crc32)是否能表决出结果，能表决出结果再做大数比较
    modeIdx_0 = FindMode(trans->data, MAX_CLIENTS, g_pconfig->N_in_NooM,
                                                            trans->dataState);
    modeIdx_1 = MajorityElement(trans->data, MAX_CLIENTS, g_pconfig->N_in_NooM,
                                                            trans->dataState);
    if ((modeIdx_0 > -1) && (modeIdx_0 == modeIdx_1)) {
        int idx_0 = BigDataVoter(trans->bigVotingData, MAX_CLIENTS,
                                        trans->lenData, trans->dataState,
                                        g_pconfig->M_in_NooM, &votingResult);
        if ((VOTING_PASS == votingResult) && (idx_0 > -1)) {
#ifdef __DBEUG__
            printf("Vote success! trans->lenData: %d, modeIdx_0: %d\n", trans->lenData, modeIdx_0);
            //print_hex((char*)(trans->bigVotingData[modeIdx_0].bytes), trans->lenData);
#endif

        }
        else {
#ifdef __DBEUG__
            printf("Vote fail(no mode)! BigDataVoter return: %d\n", idx_0);
#endif
            modeIdx_0 = -1;
        }
    }
    else {
        modeIdx_0 = -1;
#ifdef __DBEUG__
        printf("Vote fail(crc32)!\n");
#endif
    }

    // modeIdx_0 : > -1 on success, == -1 on failure
    return modeIdx_0;

} // end of Vote()

/**
 * GetMonitorInfo
 */
void GetMonitorInfo(char *buf, int sizeOfBuf)
{
    snprintf(buf, sizeOfBuf, "{\"numclients:\"%d,"
                             "\"min_clients:\"%d,"
                             "\"cur_idTransaction:\"%ld}",
                             p_Chat->numclients,
                             MIN_CLIENTS,
                             p_Chat->cur_idTransaction);
}

void StartNewVote(VoterMessage *box, VoterMessage *message, VoterClient *c, uint64_t clock)
{
    int16_t num;
    struct timespec now;

    (void)clock;

    clock_gettime(CLOCK_MONOTONIC, &now);

    // Check if the client's requests are too frequently
    int64_t time_ms = now.tv_sec * 1000000 + now.tv_nsec / 1000 -
                                  //123456                            123456
                                    (p_Chat->lastAccessTime.tv_sec * 1000000 +
                                    p_Chat->lastAccessTime.tv_nsec / 1000);
    if (time_ms < g_pconfig->minimum_VOTING_INTERVAL) {
        // Send the received packets back to the client
        if (DIALOG_CLIENT_SERVER == g_Voter_Working_Mode) {
            num = write(c->fd, box, box->len);
            if (num != box->len) {
                return;
            }
        }
#ifdef __DBEUG__
        printVoterMsg("StartNewVote (1)", box);
#endif
        return;
    }

    // 防护一下，免得数据访问越界
    if ((p_Chat->cur_trans_id < 0) || (p_Chat->cur_trans_id > MAX_NUM_TRANS)) {
        syslog(LOG_ERR, "StartNewVote, p_Chat->cur_idTransaction: Bad data");
        return;
    }

    // 检查离最近的表决事务是否太近
    Transaction *trans = &(p_Chat->trans_arr[p_Chat->cur_trans_id]);
    time_ms = now.tv_sec * 1000000 + now.tv_nsec / 1000 -
                          //123456                              123456
                                (trans->establishTime.tv_sec * 1000000 +
                                trans->establishTime.tv_nsec / 1000);

    if (time_ms < g_pconfig->minimum_VOTING_INTERVAL) {
        if (DIALOG_CLIENT_SERVER == g_Voter_Working_Mode) {
            // Send the received packets back to the client
            num = write(c->fd, box, box->len);
            if (num != box->len) {
                return;
            }
        }
#ifdef __DBEUG__
        printVoterMsg("StartNewVote (2)", box);
#endif
        return;
    }

    /**
     * Start a new vote _________________________
     */
    p_Chat->current_data_num = 0;

    // 循环使用 p_Chat->trans_arr 中的空间
    p_Chat->cur_trans_id = (p_Chat->cur_trans_id + 1) % MAX_NUM_TRANS;

    // 表示当前表决事务无数据
    p_Chat->cur_idTransaction = UINT64_MAX;

    p_Chat->last_Success_idTransaction = UINT64_MAX;

    // Set lamportClock & establishTime
    // For now we don't use lamportClock, beacuse phxpaxos
    // can do sync work as a substitute.
    // p_Chat->lamportClock[p_Chat->cur_trans_id] = clock;

    trans = &(p_Chat->trans_arr[p_Chat->cur_trans_id]);

    clock_gettime(CLOCK_REALTIME, &(trans->establishTime));

    // Clean up voting transaction data
    for (int32_t i = 0;  i < MAX_CLIENTS;  i++) {
        trans->dataState[i] = DATA_NULL;
    }
    trans->cur_data_num = 0;

    /* Send vote notices to all clients */
#ifdef __DBEUG__
    printVoterMsg("StartNewVote (3)", message);
#endif

    if (DIALOG_CLIENT_SERVER == g_Voter_Working_Mode) {
        SendMsgToAllClients(-1, (char*)message, message->len);
    }
}

int RequestVote(VoterMessage *box, VoterClient *c, char *messageBuf, uint64_t clock)
{
    VoterMessage *message;
    Int64         i64;
    int           iRet;
    int16_t       num;
    char          type;

    if (0 != g_pconfig->StrongOrder) {
        if ((UINT64_MAX != p_Chat->last_Success_idTransaction) &&
            (box->idTransaction != p_Chat->last_Success_idTransaction) &&
            (! is_next_range(box->idTransaction, p_Chat->last_Success_idTransaction,
                       MAX_NUM_TRANS, 1, g_pconfig->numTransPerVotingcycle))) {
            return 0;
        }
    }

    i64.i64Val = 0;

    // Vote condition has been checked

    // For now we don't use lamportClock
    // int trans_idx = -1;
    // for (int i = 0;  i < MAX_NUM_TRANS;  i++) {
    //     if (p_Chat->lamportClock[i] == box->idTransaction) {
    //         trans_idx = i;
    //         break;
    //     }
    // }

    // Find the corresponding transaction
    int32_t trans_idx = GetTransIdx(box->idTransaction);
#ifdef __DBEUG__
    printf("RequestVote (0), trans_idx: %d = "
           "GetTransIdx(box->idTransaction: %ld, box->len: %d)\n", trans_idx,
           box->idTransaction, box->len);
#endif
    // Didn't find the transaction you wanted to vote on
    if (-1 == trans_idx) {
        message = BoxVoterMessage(messageBuf,
                                    MAX_VoterMessage_SIZE,
                                    VOTER_RESULT_NOID,
                                    p_Chat->cur_idTransaction,
                                            // uint64_t idTransaction
                                    clock,  // uint64_t lamportClock,
                                    box->idTransaction,
                                            // int64_t data
                                    NULL,   // char *str
                                    0
                                    );

        // Send data to the client
        if (DIALOG_CLIENT_SERVER == g_Voter_Working_Mode) {
            num = write(c->fd, message, message->len);
            if (num != sizeof(VoterMessage)) {
                // In this case, sizeof(VoterMessage) == message->len
                return -1;
            }
        }
#ifdef __DBEUG__
        printVoterMsg("RequestVote (1)", message);
#endif
        return -2;
    }

    Transaction *trans = &(p_Chat->trans_arr[trans_idx]);

    // 如果本事务已经表决过，就不再重复表决了
    if (trans->votingResult == VOTING_PASS) {
        // 把本次表决看作是成功的，让后续能做表决成功后的数据清理工作。
        return 0;
    }

    int modeIdx = Vote(trans);
    if (modeIdx < 0) {
        if (-2 == modeIdx) {
            return 2;
        }

        type = VOTER_RESULT_FAIL;

        // 本表决未通过，结果为失败
        trans->votingResult = VOTING_FAIL;
    }
    else {
        type = VOTER_RESULT_RAW;
        i64 = trans->data[modeIdx];

        // 本表决已经通过，结果为成功
        trans->votingResult = VOTING_PASS;
    }

    // Package the voting results data
    trans->mode = i64;

    if (DIALOG_CLIENT_SERVER == g_Voter_Working_Mode) {
        message = BoxVoterMessage(messageBuf,
                                    MAX_VoterMessage_SIZE,
                                    type,
                                    box->idTransaction,
                                                // uint64_t idTransaction
                                    clock,      // uint64_t lamportClock,
                                    i64.i64Val, // int64_t data
                                    NULL,       // char *str
                                    0
                                    );

        // Send the voting results to all clients
#ifdef __DBEUG__
        printVoterMsg("RequestVote (2)", message);
#endif

        SendMsgToAllClients(-1, (char*)message, message->len);
    }
    else if ((MORE_IN_ONE_OUT_AT_ONCE_WITH_SYNC == g_Voter_Working_Mode) ||
                          (MORE_IN_ONE_OUT_WITH_SYNC == g_Voter_Working_Mode)) {
        if ((VOTER_RESULT_RAW == type) &&
            ((0 == g_pconfig->onlyMasterOutput) ||
             ('p' != g_pconfig->cDataSynchronizationProtocol) ||
                                        (0 == PhxIsIMMaster(g_oPaxosServer)))) {

            notify_select_exit(box->idTransaction);

            /**
             * 一个表决器集群，由多个节点组成，这些表决节点必须多数表决通过才能算总的表决通
             * 过，以此来防止节点失效，达到功能安全的目的。那种每个节点自己表决通过后，把表
             * 决结果通过PhxPaxos同步出去，真要对外输出表决结果时，要判断是不是大多数的表
             * 决器都表决通过。这中间要有一个延迟。
             *
             * 只有第一次表决成功，才等其它表决器的表决结果。
             */
            int64_t delayTime;
            int32_t i = 0;
            do {
                iRet = ClusterCheckVoterResult(box);
                if (0 == iRet) {
                    break;
                }
                (void)TimespecWaitNanoseconds(NULL, &delayTime,
                                              NANOSECONDS_FOR_SCHEDULE * 3);

                i++;
            } while (i < g_pconfig->maxWait);

            if (i >= g_pconfig->maxWait) {
                // 表决未通过
#ifdef __DBEUG__
                printf("ClusterCheckVoterResult failed. iRet: %d\n", iRet);
#endif
                syslog(LOG_ERR, "ClusterCheckVoterResult failed. iRet: %d", iRet);
                return 0;
            }

#ifdef __DBEUG__
            printf(">>>>>>>> ClusterCheckVoterResult(box): %d\n", iRet);
#endif
            // output voting result

            /**
             * modeIdx 这时是表决通过的数据的最靠前的那个的下标。
             */
            g_voterOutputTokenOrder = 3;
            if (NULL != p_Chat->pVoterOutputMsg) {
                iRet = p_Chat->pVoterOutputMsg((char*)trans->bigVotingData[modeIdx].bytes,
                                               trans->lenData, 3, VOTER_TOKEN_3, trans);
            }
            else {
                iRet = VoterOutputMsg((char*)trans->bigVotingData[modeIdx].bytes,
                                      trans->lenData, 3, VOTER_TOKEN_3, trans);
            }

            if (iRet < 0) {
                g_pVoterRunningState->votingResultFail++;
                WriteLog("error, voting_voter_main", -1);
            }
        }
    }

    return 0;
}

/**
 * 数据包分析，消息处理
 */
int ProcessProtocol(VoterMessage *box, int bufSize, int clientId, bool needSync)
{
    // i64 is temporarily used to store message data
    Int64         i64;
    char          messageBuf[MAX_VoterMessage_SIZE];
    VoterMessage *message;
    int16_t       num = 0;
    uint64_t      clock;
    int           iRet;

    /**
     * Voter 工作在模式： ONE_IN_ONE_OUT_WITH_SYNC
     */
    if ((ONE_IN_ONE_OUT_WITH_SYNC == g_Voter_Working_Mode) &&
                                           (0 != g_pconfig->onlyMasterOutput) &&
             (('p' != g_pconfig->cDataSynchronizationProtocol) ||
                                        (0 == PhxIsIMMaster(g_oPaxosServer)))) {
        if (NULL != p_Chat->pVoterOutputMsg) {
            iRet = p_Chat->pVoterOutputMsg((char*)box, bufSize, 3, VOTER_TOKEN_3, NULL);
        }
        else {
            iRet = VoterOutputMsg((char*)box, bufSize, 3, VOTER_TOKEN_3, NULL);
        }

        if (iRet < 0) {
            g_pVoterRunningState->votingResultFail++;
            WriteLog("error, voting_voter_main", -1);
        }

        return 0;
    }

    if (NULL == CheckBoxVoterMessage(box, bufSize)) {
        // The received packet validation failed
        syslog(LOG_ERR, "ProcessProtocol: Bad packet");
#ifdef __DBEUG__
        printVoterMsg("ProcessProtocol: Bad packet", box);
#endif
        return -1;
    }

#ifdef SYNC_WITH_PHXPAXOS
    if ((NULL != g_oPaxosServer) && needSync) {
        int32_t action = 0x1111;    // ComoPhxUtils::LevelDbWrite;
        char bufTmp[1];
        size_t bufTmpSize = 0;

        VoterClient *c = &(p_Chat->clients[clientId]);

        char   idVoterBuf[MAX_ID_SIZE + 128];
        snprintf(idVoterBuf, sizeof(idVoterBuf), MACRO_VOTERID_PRINTF_FMTSTR_2,
                                 g_pconfig->id, c->idVoter, box->idTransaction);

        // 数据同步协议：paxos 或 ZMQ，缺省 ZMQ
        if ('p' == g_pconfig->cDataSynchronizationProtocol) {
            iRet = PhxPropose(g_oPaxosServer, action, idVoterBuf,
                                     (char *)box, bufSize, bufTmp, &bufTmpSize);
        }
        else {
            iRet = PhxProposeDirectly(g_oPaxosServer, action, idVoterBuf,
                                                          (char *)box, bufSize);
        }
        if (0 != iRet) {
            syslog(LOG_ERR, "PhxPropose failed. iRet: %d", iRet);
            return -1;
        }
    }
    else {
#ifdef __DBEUG__
        printVoterMsg("ProcessProtocol (0)", box);
#endif
    }

#else
    (void)needSync;
#endif

    pthread_mutex_lock(&(p_Chat->mutex));

    VoterClient *c = &(p_Chat->clients[clientId]);
    clock = local_event(&g_LamportClock);

    switch (box->type) {
        // Type 1: Synchronize the clock (by clients)
        case VOTER_SYNC_CLOCK:
            message = BoxVoterMessage(messageBuf,
                                        MAX_VoterMessage_SIZE,
                                        VOTER_SYNC_CLOCK,
                                        box->idTransaction,
                                                    // uint64_t idTransaction
                                        clock,      // uint64_t lamportClock,
                                        clock,      // int64_t data
                                        box->buf,   // char *str
                                        0
                                     );
            // Send data to the client
            if (DIALOG_CLIENT_SERVER == g_Voter_Working_Mode) {
                num = write(c->fd, message, message->len);
                if (num != sizeof(VoterMessage)) {
                    // In this case, sizeof(VoterMessage) == message->len
                    break;
                }
            }
#ifdef __DBEUG__
            printVoterMsg("ProcessProtocol (1)", message);
#endif
            break;

        // Type 2: Monitor clients & current trans
        case VOTER_CMD_MONITOR: {
            char buf[MAX_VoterMessage_BUF_SIZE];

            GetMonitorInfo(buf, MAX_VoterMessage_BUF_SIZE);

            // Package the data to be sent to the client
            message = BoxVoterMessage(messageBuf,
                                        (int32_t)MAX_VoterMessage_SIZE,
                                        VOTER_RESULT_RAW,
                                        p_Chat->cur_idTransaction,
                                                    // uint64_t idTransaction
                                        0,          // uint64_t lamportClock,
                                        0,          // int64_t data
                                        buf,        // char *str
                                        0
                                     );

            if (DIALOG_CLIENT_SERVER == g_Voter_Working_Mode) {
                // Send data to the client
                num = write(c->fd, message, message->len);
                if (num != message->len) {
                    break;
                }
            }
#ifdef __DBEUG__
            printVoterMsg("ProcessProtocol (2)", message);
#endif
            break;
        }

        // Type 3: Start a new vote
        case VOTER_CMD_START_NEW: {
            message = BoxVoterMessage(messageBuf,
                                        MAX_VoterMessage_SIZE,
                                        VOTER_SYNC_CLOCK,
                                        box->idTransaction,
                                                    // uint64_t idTransaction
                                        clock,      // uint64_t lamportClock,
                                        clock,      // int64_t data
                                        box->buf,   // char *str
                                        0
                                     );
            StartNewVote(box, message, c, clock);
            break;
        }

        // Type 4: Submit voting data (by clients)
        case VOTER_CMD_VOTING_DATA:
            // Extract transaction data
            // ExtractData() stores the data into i64Val
            i64.i64Val = 0;
            iRet = ExtractData(clientId, box);
            if (0 != iRet) {
                syslog(LOG_ERR, "ExtractData failed. iRet: %d", iRet);
                break;
            }

            // Package the data to be sent to the client
            message = BoxVoterMessage(messageBuf,
                                        MAX_VoterMessage_SIZE,
                                        VOTER_RESULT_RAW,
                                        p_Chat->cur_trans_id,
                                                    // uint64_t idTransaction
                                        clock,      // uint64_t lamportClock,
                                        i64.i64Val, // int64_t data
                                        NULL,       // char *str
                                        0
                                     );

            // p_Chat->numclients 会在 ExtractData() 中改变
            // 如果有足够多的数据就表决
            if ((p_Chat->numclients >= g_pconfig->M_in_NooM) &&
                               (p_Chat->current_data_num >= g_pconfig->N_in_NooM)) {
                if (RequestVote(box, c, messageBuf, clock) == 0) {
                    // 即使表决不成功，也要开始新的表决
                    StartNewVote(box, message, c, clock);
                }
            }

            // Send data to the client
            if (DIALOG_CLIENT_SERVER == g_Voter_Working_Mode) {
                // Send data to the client
                num = write(c->fd, message, message->len);
                if (num != message->len) {
                    break;
                }
            }
#ifdef __DBEUG__
            printVoterMsg("ProcessProtocol (3)", message);
#endif
            break;

        // Type 5: Submit big voting data (by clients)
        case VOTER_CMD_VOTING_BIGDATA:
            // Extract transaction data
            // ExtractData() stores the data into i64Val
            i64.i64Val = 0;

            iRet = ExtractBigData(clientId, box);
            if (0 != iRet) {
#ifdef __DBEUG__
                printf("ExtractBigData failed. iRet: %d\n", iRet);
#endif
                syslog(LOG_ERR, "ExtractBigData failed. iRet: %d", iRet);
                break;
            }

            int str_len = box->len - VoterMessage_HEAD_LEN;
            // Package the data to be sent to the client
            message = BoxVoterMessage(messageBuf,
                                        MAX_VoterMessage_SIZE,
                                        VOTER_RESULT_RAW,
                                        p_Chat->cur_trans_id,
                                                    // uint64_t idTransaction
                                        clock,      // uint64_t lamportClock,
                                        i64.i64Val, // int64_t data
                                        box->buf,   // char *str
                                        str_len
                                     );
#ifdef __DBEUG__
            printf("=ProcessProtocol=> p_Chat->numclients: %d, g_pconfig->M_in_NooM: %d, "
                   "p_Chat->current_data_num: %d, g_pconfig->N_in_NooM: %d\n",
                   p_Chat->numclients, g_pconfig->M_in_NooM,
                   p_Chat->current_data_num, g_pconfig->N_in_NooM);
#endif
            // There must be enough clients and data for vote.
            /**
             * 如果一个表决周期内要表决多个事务，这个是至少满足的条件
             * (p_Chat->current_data_num >= g_pconfig->N_in_NooM * MAX_NUM_TRANS)
             */

            if ((p_Chat->numclients >= g_pconfig->M_in_NooM) &&
                (p_Chat->current_data_num >= (g_pconfig->N_in_NooM * MAX_NUM_TRANS))) {
                if (RequestVote(box, c, messageBuf, clock) == 0) {
                    // 即使表决不成功，也要开始新的表决
                    StartNewVote(box, message, c, clock);
                }
            }
            
            if (DIALOG_CLIENT_SERVER == g_Voter_Working_Mode) {
                // Send data to the client
                num = write(c->fd, message, message->len);
                if (num != sizeof(VoterMessage)) {
                    // 这时，sizeof(VoterMessage) == message->len
                    break;
                }
            }
#ifdef __DBEUG__
            printVoterMsg("ProcessProtocol (4)", message);
#endif
            break;

        // Type 6: Ask for a vote (by clients)
        case VOTER_CMD_VOTE: {
            message = BoxVoterMessage(messageBuf,
                                        MAX_VoterMessage_SIZE,
                                        VOTER_SYNC_CLOCK,
                                        box->idTransaction,
                                                    // uint64_t idTransaction
                                        clock,      // uint64_t lamportClock,
                                        clock,      // int64_t data
                                        box->buf,   // char *str
                                        0
                                     );
            RequestVote(box, c, messageBuf, clock);
            break;
        }

        default:
            syslog(LOG_ERR, "ProcessProtocol: Bad packet");
            break;
    } // end of switch (box->type)

    if (MORE_IN_ONE_OUT_AT_ONCE_WITH_SYNC == g_Voter_Working_Mode) {
        if ((p_Chat->numclients >= g_pconfig->M_in_NooM) &&
                           (p_Chat->current_data_num >= g_pconfig->N_in_NooM)) {
            Transaction *trans;

            // 防护一下，免得数据访问越界
            if ((p_Chat->cur_trans_id < 0) || (p_Chat->cur_trans_id > MAX_NUM_TRANS)) {
                pthread_mutex_unlock(&(p_Chat->mutex));

                syslog(LOG_ERR, "ExtractData, p_Chat->cur_idTransaction: Bad data");
                return -1;
            }

            trans = &(p_Chat->trans_arr[p_Chat->cur_trans_id]);
            /**
             * 虽然当前的表决不是周期事务（即一个周期内的所有表决事务，必须同时成功，只要有一个
             * 不成功，整体不输出），但因为表决缓存池的原因，也需要保持其有序，只是表决不成功的
             * 事务可以丢掉。
             */
            if ((VOTING_PASS != trans->votingResult) &&
                                    (try_to_process_vote(0) >= MAX_NUM_TRANS)) {
                pthread_mutex_unlock(&(p_Chat->mutex));

                // 本表决已经通过，结果为成功
                trans->votingResult = VOTING_PASS;
                p_Chat->last_Success_idTransaction = trans->idTransaction;
                p_Chat->current_data_num = 0;

                return 0;
            }

            // 本表决未通过，继续接收数据
        }
    }
    else if (MULTI_IN_ONE_BY_ONE_OUT_PERIODIC == g_Voter_Working_Mode) {
        if (p_Chat->numclients >= g_pconfig->M_in_NooM) {
            int i;
            Transaction *trans = NULL;

            for (i = 0;  i < MAX_NUM_TRANS;  i++) {
                /**
                 * 如果是强有序表决，try_to_process_vote 实际没表决，现在发现可以表决成功，
                 * 于是真的做表决
                 */
                trans = &(p_Chat->trans_arr[i]);
                if (trans->cur_data_num < g_pconfig->N_in_NooM) {
                    // data not ready
                    break;
                }
            }

            if (i >= MAX_NUM_TRANS) {
                if (try_to_process_vote(g_pconfig->StrongOrder) >= MAX_NUM_TRANS) {
                    /**
                     * 如果是强有序表决，try_to_process_vote 实际没表决，现在发现可以表决成功，
                     * 于是真的做表决
                     */
                    if (g_pconfig->StrongOrder) {
                        try_to_process_vote(1);
                    }

                    pthread_mutex_unlock(&(p_Chat->mutex));

                    // 本表决已经通过，结果为成功
                    if (NULL != trans) {
                        p_Chat->last_Success_idTransaction = trans->idTransaction;
                    }
                    p_Chat->current_data_num = 0;

                    return 0;
                }
            }
        }
    }

    pthread_mutex_unlock(&(p_Chat->mutex));

    return 0;
} // end of ProcessProtocol()
